package org.micro.neural.config;

import com.google.common.util.concurrent.ThreadFactoryBuilder;
import org.micro.neural.URL;
import org.micro.neural.common.Constants;
import org.micro.neural.common.event.EventProcessor;
import org.micro.neural.config.store.IStore;
import org.micro.neural.config.store.IStoreListener;
import org.micro.neural.config.store.ModuleKey;
import org.micro.neural.config.store.StoreKey;
import org.micro.neural.extension.Extension;
import org.micro.neural.extension.ExtensionLoader;
import org.micro.neural.extension.SPI;
import lombok.extern.slf4j.Slf4j;

import java.io.Serializable;
import java.lang.reflect.ParameterizedType;
import java.lang.reflect.Type;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.*;

/**
 * The Store Config
 *
 * @param <C> extends {@link NeuralConfig}
 * @param <G> extends {@link GlobalConfig}
 * @author lry
 **/
@SPI
@Slf4j
public class StoreConfig<C extends NeuralConfig, G extends GlobalConfig> implements Serializable {

    private final Class<C> configClass;
    private final Class<G> globalClass;

    private IStore store;

    protected final String module;
    protected volatile G globalConfig;
    protected volatile boolean enable = true;
    protected volatile ConcurrentMap<String, C> configs = new ConcurrentHashMap<>();

    private IStoreListener listener = null;
    private ExecutorService subscribeConfigExecutor = null;
    private ScheduledExecutorService pullConfigExecutor = null;
    private ScheduledExecutorService pushStatisticsExecutor = null;

    @SuppressWarnings("unchecked")
    public StoreConfig() {
        this.module = this.getClass().getAnnotation(Extension.class).value();

        Type type = this.getClass().getGenericSuperclass();
        Type[] args = ((ParameterizedType) type).getActualTypeArguments();
        this.configClass = (Class<C>) args[0];
        this.globalClass = (Class<G>) args[1];
    }

    public void setEnable(boolean enable) {
        this.enable = enable;
    }

    /**
     * The add degrade
     *
     * @param config {@link C}
     */
    public void addConfig(C config) {
        configs.put(config.identity(), config);
    }

    /**
     * The initialize
     *
     * @param url {@link URL}
     */
    public void initialize(URL url) {
        if (!enable) {
            log.warn("The limiter already is started");
            return;
        }
        if (configs.isEmpty()) {
            throw new RuntimeException("The config list is empty.");
        }

        this.store = ExtensionLoader.getLoader(IStore.class).getExtension(url.getProtocol());
        store.initialize(url);

        // add limiter global config data to remote
        G remoteGlobalConfig = store.query(new ModuleKey(module), new StoreKey(Constants.GLOBAL_CONFIG), globalClass);
        if (null == remoteGlobalConfig) {
            try {
                store.add(new ModuleKey(module), new StoreKey(Constants.GLOBAL_CONFIG), globalClass.newInstance());
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
        // add limiter config data to remote
        configs.forEach(((identity, config) -> {
            C remoteConfig = store.query(new ModuleKey(module), new StoreKey(identity), configClass);
            if (null == remoteConfig) {
                store.add(new ModuleKey(module), new StoreKey(identity), config);
            }
        }));

        // initialize pull all configs
        this.pullConfigs();
        // start cycle pull configs
        this.cyclePullConfigs();
        // start subscribe configs
        this.subscribeConfigs();
        // start cycle push statistics
        this.cyclePushStatistics();

        // add shutdown Hook
        Runtime.getRuntime().addShutdownHook(new Thread(this::destroy));
    }

    /**
     * The notify changed config
     *
     * @param identity the config identity
     * @param config   the config
     */
    protected void notify(String identity, C config) {

    }

    /**
     * The collect statistics data
     *
     * @return statistics data
     */
    protected Map<String, Long> collect() {
        return null;
    }

    /**
     * The destroy store config
     */
    public void destroy() {
        log.debug("The {} is executing destroy……", module);
        if (null != pullConfigExecutor) {
            pullConfigExecutor.shutdown();
        }
        if (null != subscribeConfigExecutor) {
            subscribeConfigExecutor.shutdown();
        }
        if (null != pushStatisticsExecutor) {
            pushStatisticsExecutor.shutdown();
        }
        this.unSubscribe(listener);
        if (null != store) {
            store.destroy();
        }
    }

    /**
     * The pull all configs
     */
    private void pullConfigs() {
        // cycle pull global config
        try {
            G remoteGlobalConfig = store.query(new ModuleKey(module), new StoreKey(Constants.GLOBAL_CONFIG), globalClass);
            if (null != remoteGlobalConfig && !remoteGlobalConfig.equals(globalConfig)) {
                log.debug("The {} global config pull changed: {}", module, remoteGlobalConfig);
                this.globalConfig = remoteGlobalConfig;
            }
        } catch (Exception e) {
            EventProcessor.EVENT.notify(module, EventType.PULL_GLOBAL_CONFIG_EXCEPTION);
            log.error("The " + module + " cycle pull global config is exception", e);
        }

        // cycle pull resource config
        configs.forEach((identity, config) -> {
            try {
                C remoteConfig = store.query(new ModuleKey(module), new StoreKey(identity), configClass);
                if (null != remoteConfig && !remoteConfig.equals(config)) {
                    log.debug("The {} config pull changed: {}, {}", module, identity, remoteConfig);
                    configs.put(identity, remoteConfig);
                    this.notify(identity, remoteConfig);
                }

                // the first excuse
                this.notify(identity, remoteConfig);
            } catch (Exception e) {
                EventProcessor.EVENT.notify(module, EventType.PULL_CONFIG_EXCEPTION);
                log.error("The " + module + " cycle pull config[" + identity + "] is exception", e);
            }
        });
    }

    /**
     * The cycle pull configs
     */
    private synchronized void cyclePullConfigs() {
        if (null != pullConfigExecutor) {
            log.warn("The {} cyclePullConfigs is executed", module);
            return;
        }

        // start pull config data executor
        log.debug("The {} executing pull config data executor……", module);
        // build Task Name
        String pullConfigName = String.join(Constants.SEQ, module, Constants.PULL_CONFIG);
        ThreadFactoryBuilder pullBuilder = new ThreadFactoryBuilder();
        ThreadFactory pullThreadFactory = pullBuilder.setDaemon(true).setNameFormat(pullConfigName).build();
        this.pullConfigExecutor = Executors.newScheduledThreadPool(1, pullThreadFactory);

        // execute schedule pull config by fixed rate
        this.pullConfigExecutor.scheduleAtFixedRate(this::pullConfigs,
                globalConfig.getPullConfigCycle(), globalConfig.getPullConfigCycle(), TimeUnit.MILLISECONDS);
    }

    /**
     * The subscribe configs
     */
    private synchronized void subscribeConfigs() {
        if (null != subscribeConfigExecutor) {
            log.warn("The {} subscribeConfigs is executed", module);
            return;
        }

        // start subscribe config data executor
        log.debug("The {} executing subscribe config data executor……", module);
        String subscribeConfigName = String.join(Constants.SEQ, module, Constants.SUBSCRIBE_CONFIG);
        ThreadFactoryBuilder subscribeBuilder = new ThreadFactoryBuilder();
        ThreadFactory subscribeThreadFactory =
                subscribeBuilder.setDaemon(true).setNameFormat(subscribeConfigName).build();
        this.subscribeConfigExecutor = Executors.newFixedThreadPool(1, subscribeThreadFactory);

        // execute subscribe global config
        this.subscribeConfigExecutor.execute(() -> {
            try {
                // the execute subscribe
                this.subscribe(module, configs.keySet(), listener = (channel, message) -> {
                    log.debug("The {} global config subscribed changed: {}, {}", module, channel, message);
                    if (null == channel || null == message) {
                        return;
                    }

                    if (channel.endsWith(Constants.SUBSCRIBE_GLOBAL_CONFIG_CHANNEL)) {
                        // the global config notify
                        globalConfig = Constants.deserialize(globalClass, message);
                    } else {
                        // the config notify
                        String[] keyArray = channel.split(Constants.DELIMITER);
                        if (keyArray.length == 5) {
                            C config = Constants.deserialize(configClass, message);
                            config.setApplication(keyArray[2]);
                            config.setGroup(keyArray[3]);
                            config.setResource(keyArray[4]);
                            configs.put(config.identity(), config);
                        }
                    }
                });
            } catch (Exception e) {
                EventProcessor.EVENT.notify(module, EventType.SUBSCRIBE_CONFIG_EXCEPTION);
                log.error("The " + module + " subscribed configs or global config is exception", e);
            }
        });
    }

    /**
     * The cycle push statistics
     */
    private synchronized void cyclePushStatistics() {
        if (null != pushStatisticsExecutor) {
            log.warn("The {} cyclePushStatistics is executed", module);
            return;
        }

        // start push statistics data executor
        log.debug("The {} executing push statistics data executor……", module);
        String pushStatisticsName = String.join(Constants.SEQ, module, Constants.PUSH_STATISTICS);
        ThreadFactoryBuilder pushBuilder = new ThreadFactoryBuilder();
        ThreadFactory pushTreadFactory = pushBuilder.setDaemon(true).setNameFormat(pushStatisticsName).build();
        this.pushStatisticsExecutor = Executors.newScheduledThreadPool(1, pushTreadFactory);

        // execute schedule push statistics by fixed rate
        this.pushStatisticsExecutor.scheduleAtFixedRate(() -> {
            try {
                // query memory statistics data
                Map<String, Long> statisticsData = this.collect();
                log.debug("The {} cycle push statistics: {}", module, statisticsData);
                if (null == statisticsData || statisticsData.isEmpty()) {
                    return;
                }

                Map<String, Long> sendData = new HashMap<>();
                for (Map.Entry<String, Long> entry : statisticsData.entrySet()) {
                    sendData.put(String.join(Constants.DELIMITER, module, entry.getKey()), entry.getValue());
                }

                // push statistics data to remote
                store.batchUpOrAdd(globalConfig.getStatisticDataExpire(), sendData);
            } catch (Exception e) {
                EventProcessor.EVENT.notify(module, EventType.PUSH_STATISTICS_EXCEPTION);
                log.error("The " + module + " cycle push statistics is exception", e);
            }
        }, globalConfig.getStatisticReportCycle(), globalConfig.getStatisticReportCycle(), TimeUnit.MILLISECONDS);
    }

    /**
     * The subscribe
     *
     * @param identity {@link NeuralConfig}
     * @param object   object
     */
    public void publish(String identity, Object object) {
        String channel = null;
        if (object instanceof GlobalConfig) {
            channel = String.join(Constants.DELIMITER, module, Constants.SUBSCRIBE_GLOBAL_CONFIG_CHANNEL);
        } else if (object instanceof NeuralConfig) {
            channel = String.format(String.join(
                    Constants.DELIMITER, module, Constants.SUBSCRIBE_CONFIG_CHANNEL_KEY), identity);
        }
        if (null == channel) {
            return;
        }

        store.publish(new ModuleKey(module), new StoreKey(channel), object);
    }

    /**
     * The subscribe
     *
     * @param module     module
     * @param identities {@link NeuralConfig} list
     * @param listener   {@link IStoreListener}
     */
    private void subscribe(String module, Set<String> identities, IStoreListener listener) {
        // the build subscribe keys
        Set<StoreKey> subscribeKeys = new HashSet<>(identities.size() + 1);
        subscribeKeys.add(new StoreKey(
                String.join(Constants.DELIMITER, module, Constants.SUBSCRIBE_GLOBAL_CONFIG_CHANNEL)));
        String configKey = String.join(Constants.DELIMITER, module, Constants.SUBSCRIBE_CONFIG_CHANNEL_KEY);
        for (String identity : identities) {
            subscribeKeys.add(new StoreKey(String.format(configKey, identity)));
        }

        // the execute subscribe
        store.subscribe(new ModuleKey(module), subscribeKeys.toArray(new StoreKey[0]), listener);
    }

    /**
     * The un subscribe
     *
     * @param listener {@link IStoreListener}
     */
    private void unSubscribe(IStoreListener listener) {
        store.unSubscribe(listener);
    }

}
